package kafkaex

import (
	"context"
	"kafkademo/internal/domain"
	"kafkademo/internal/entity"
	"kafkademo/internal/po"
	"sync"

	"github.com/IBM/sarama"
	"github.com/google/uuid"
)

func (g *ConsumerGroup) GetGroupID() string {
	return g.groupID
}

type IConsumer interface {
	IConsumerGroup
	sarama.ConsumerGroupHandler
	Go() (func(), error)
}

// IConsumer Impl
var _ = IConsumer(&Consumer{})

type Consumer struct {
	id string
	ConsumerGroup
	topics []string
	ready  chan bool
}

func (c *Consumer) Go() (func(), error) {
	ctx, cancel := context.WithCancel(context.Background())
	wg := &sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			if err := c.group.Consume(ctx, c.topics, c); err != nil {
				// 当setup失败的时候，error会返回到这里
				logger.Error(ctx, "Error from consumer: %v", err)
				return
			}
			// check if context was cancelled, signaling that the consumer should stop
			if ctx.Err() != nil {
				logger.Println(ctx.Err())
				return
			}
			c.ready = make(chan bool)
		}
	}()
	<-c.ready
	logger.Info(ctx, "Sarama consumer up and running!...")
	// 保证在系统退出时，通道里面的消息被消费
	return func() {
		logger.Println("kafka close")
		cancel()
		wg.Wait()
		if err := c.group.Close(); err != nil {
			logger.Printf("Error closing client: %v \n", err)
		}
	}, nil
}

// Setup is run at the beginning of a new session, before ConsumeClaim
func (c *Consumer) Setup(sarama.ConsumerGroupSession) error {
	// Mark the consumer as ready
	close(c.ready)
	return nil
}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
// Once the Messages() channel is closed, the Handler must finish its processing
// loop and exit.
func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	d := domain.NewMessageDomain()
	// NOTE:
	// Do not move the code below to a goroutine.
	// The `ConsumeClaim` itself is called within a goroutine, see:
	// https://github.com/IBM/sarama/blob/main/consumer_group.go#L27-L29
	for {
		select {
		case message, ok := <-claim.Messages():
			if !ok {
				logger.Printf("message channel was closed")
				return nil
			}
			d.Save(session.Context(), &entity.MqMessage{
				MqMessage: po.MqMessage{
					Id:      uuid.NewString(),
					Key:     string(message.Key),
					Topic:   message.Topic,
					Raw:     string(message.Value),
					Ts:      message.Timestamp.Unix(),
					BlockTs: message.BlockTimestamp.Unix(),
				},
			})
			logger.Printf("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
			session.MarkMessage(message, "")
		// Should return when `session.Context()` is done.
		// If not, will raise `ErrRebalanceInProgress` or `read tcp <ip>:<port>: i/o timeout` when kafka rebalance. see:
		// https://github.com/IBM/sarama/issues/1192
		case <-session.Context().Done():
			return nil
		}
	}
}
